package net.qiujuer.library.clink.impl.async;

import net.qiujuer.library.clink.core.IoArgs;
import net.qiujuer.library.clink.core.SendDispatcher;
import net.qiujuer.library.clink.core.SendPacket;
import net.qiujuer.library.clink.core.Sender;
import net.qiujuer.library.clink.utils.CloseUtils;

import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncSendDispatcher implements SendDispatcher, IoArgs.IoArgsEventProcessor {
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final Sender sender;
    private final Queue<SendPacket> queue = new ConcurrentLinkedQueue<>();
    // 发送过程当前有数据来的时候，我们要Packet丢到queue当中去，然后从queue当中把每一个Packet拿出来发送，而如果说当前正在
    // 进行一个发送状态。当咋们没有数据发送的时候，也就是说我们不再去监听发送的空闲状态，这个时候我们的发送状态是一个关闭状态
    // 知道有新数据来的时候，我们会重新去激活这个状态，所以我们需要一个变量来维护这个状态。
    private final AtomicBoolean isSending = new AtomicBoolean();

    private IoArgs ioArgs = new IoArgs();
    private SendPacket<?> packetTemp;

    private ReadableByteChannel packetChannel;
    // 因为Packet有可能会比IoArgs更大，所以需要一个进度来维护他，这里定义一个当前Packet最大的值。
    private long total;
    // 当前Packet发送了多长
    private long position;

    public AsyncSendDispatcher(Sender sender) {
        this.sender = sender;
        sender.setSendListener(this);
    }

    @Override
    public void send(SendPacket packet) {
        queue.offer(packet);
        // 如果当前不在发送中，就设置为发送k中
        if (isSending.compareAndSet(false,true)){
            sendNextPacket();
        }
    }

    private SendPacket takePacket() {
        SendPacket packet = queue.poll();
        if (packet != null && packet.isCanceled()) {
            // 已取消，不用发送，递归拿下一条
            return takePacket();
        }
        return packet;
    }

    // 发送一条数据
    private void sendNextPacket() {
        SendPacket temp = packetTemp;
        // 我们要发送下一条的时候，发现我们当前的发送的这条还不等于空
        if (temp!=null) {
            CloseUtils.close(temp);
        }

        SendPacket packet = takePacket();
        if (packet == null) {
            // 队列为空，取消状态发送
            isSending.set(false);
            return;
        }

        total = packet.length();
        position = 0;
        sendCurrentPacket();
    }

    private void sendCurrentPacket() {
        if (position>=total) {
            completePacket(position == total);
            sendNextPacket();
            return;
        }

        try {
            // 如果没有发送完成，接着发送下一个
            sender.postSendASync();
        } catch (IOException e) {
            closeAndNotify();
        }

    }

    /**
     * 完成Packet发送
     * @param isSucceed 是否成功
     */
    private void completePacket(boolean isSucceed) {
        SendPacket packet = this.packetTemp;
        if (packet == null) {
            return;
        }
        CloseUtils.close(packet);
        CloseUtils.close(packetChannel);

        packetTemp = null;
        packetChannel = null;
        total = 0;
        position = 0;
    }

    private void closeAndNotify() {
        CloseUtils.close(this);
    }

    @Override
    public void cancel(SendPacket packet) {

    }

    @Override
    public void close() throws IOException {
        if (isClosed.compareAndSet(false, true)) {
            isSending.set(false);
            // 异常关闭导致的完成操作
            completePacket(false);
        }
    }

    @Override
    public IoArgs provideIoArgs() {
        IoArgs args = ioArgs;
        if (packetChannel == null) {
            packetChannel = Channels.newChannel(packetTemp.open());
            args.limit(4);
            args.writeLength((int) packetTemp.length());
        }else {
            args.limit((int) Math.min(args.capacity(), total - position));

            try {
                int count = args.readFrom(packetChannel);
                position += count;
            } catch (IOException e) {
                e.printStackTrace();
                return null;
            }
        }

        return args;
    }

    @Override
    public void onConsumeFailed(IoArgs args, Exception e) {
        e.printStackTrace();
    }

    @Override
    public void onConsumeCompleted(IoArgs args) {
        // 继续发送当前包
        sendCurrentPacket();
    }
}
